package com.atguigu.state;

import com.atguigu.bean.WaterSensor;
import com.atguigu.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

// TODO 水位超过指定的阈值发送告警，阈值可以动态修改
public class BroadcastStateDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);

        // 数据流
        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("10.75.186.206", 9999)
                .map(new WaterSensorMapFunction());

        // 水位值配置流
        DataStreamSource<String> configStream = env.socketTextStream("10.75.186.206", 9998);

        // TODO 1. 将 配置流 广播
        MapStateDescriptor<String, Integer> mapStateDescriptor = new MapStateDescriptor<>("mapStateDescriptor", String.class, Integer.class);
        BroadcastStream<String> broadcastStream = configStream.broadcast(mapStateDescriptor);

        // TODO 2.把 数据流 和 广播后的配置流 connect
        sensorDS.connect(broadcastStream)
                // TODO 3.调用 process
                .process(new BroadcastProcessFunction<WaterSensor, String, String>() {
                    @Override
                    public void processElement(WaterSensor value, BroadcastProcessFunction<WaterSensor, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        // TODO 5.通过上下文获取广播状态，取出里面的值（只读，不能修改）
                        ReadOnlyBroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        Integer threshold = broadcastState.get("threshold");
                        if (threshold == null) {
                            threshold = 0;
                        }
                        if (value.getVc() > threshold) {
                            out.collect(value.toString() + "水位线超过阈值：" + threshold);
                        }
                    }

                    @Override
                    public void processBroadcastElement(String value, BroadcastProcessFunction<WaterSensor, String, String>.Context ctx, Collector<String> out) throws Exception {
                        // TODO 4. 通过上下文获取广播状态，往里面写数据
                        BroadcastState<String, Integer> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
                        broadcastState.put("threshold", Integer.parseInt(value));
                        System.out.println("阈值调整为" + value);
                    }
                })
                .print();

        env.execute();
    }

}
